package com.flinksql.test;

import com.flinksql.bean.WaterSensor;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.GroupedTable;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author: Lin
 * @create: 2021-06-16 10:21
 * @description: FlinkTableAPI聚合操作
 **/
public class FlinkTableAPI_Test2 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("hadoop102", 9999);
        SingleOutputStreamOperator<WaterSensor> mapDS = source.map(new RichMapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0]
                        , Long.parseLong(split[1])
                        , Integer.parseInt(split[2]));
            }
        });

        //1.创建表的执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //2.创建表：将流转换成动态表。
        Table table = tableEnv.fromDataStream(mapDS);
        //3.对动态表进行查询，注意先后顺序，where groupby aggregate select
        Table selectTbale = table.where($("vc").isGreaterOrEqual(20))
                .groupBy($("id"))
                .aggregate($("vc").sum().as("vcSum"))
                .select($("id"), $("vcSum"));
        //4.把动态表装换成流，使用通过格式row获取
        DataStream<Tuple2<Boolean, Row>> dataStreamDS = tableEnv.toRetractStream(selectTbale, Row.class);

        dataStreamDS.print();

        env.execute();
    }
}
